package com.demo.kafka;

import java.io.UnsupportedEncodingException;
import java.util.List;
import java.util.Properties;

import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * kafka订购者
 *
 */
public abstract class KafkaSubscriber extends Thread {
	//初始化logger
	public static final Logger logger = LoggerFactory.getLogger(KafkaSubscriber.class);
	//地址
	private String address;
	//组别id
	private String groupId;
	//自动提交
	private boolean autoCommit;
	//批量长度
	private int batchSize;
	//暂停时间
	private long timeout;
	//串行键
	private String keyDeserializer;
	//串行值
	private String valueDeserializer;
	//主题
	private List<String> topics;
	//属性
	private Properties props;
	//消费者
	protected KafkaConsumer<String, byte[]> consumer;
	//是否正在运行
	protected boolean running = false;


	/**
	 * 获取address
	 * @return the address
	 */
	public String getAddress() {
		return address;
	}

	/**
	 * 设置address
	 * @param address the address to set
	 */
	public void setAddress(String address) {
		this.address = address;
	}

	/**
	 * 获取groupId
	 * @return the groupId
	 */
	public String getGroupId() {
		return groupId;
	}

	/**
	 * 设置groupId
	 * @param groupId the groupId to set
	 */
	public void setGroupId(String groupId) {
		this.groupId = groupId;
	}

	/**获取是否自动提交
	 * @return the autoCommit
	 */
	public boolean isAutoCommit() {
		return autoCommit;
	}

	/**
	 * 设置是否自动提交
	 * @param autoCommit the autoCommit to set
	 */
	public void setAutoCommit(boolean autoCommit) {
		this.autoCommit = autoCommit;
	}

	/**
	 * 获取batchSize
	 * @return the batchSize
	 */
	public int getBatchSize() {
		return batchSize;
	}

	/**
	 * 设置batchSize
	 * @param batchSize the batchSize to set
	 */
	public void setBatchSize(int batchSize) {
		this.batchSize = batchSize;
	}

	/**
	 * 获取timeout
	 * @return the timeout
	 */
	public long getTimeout() {
		return timeout;
	}

	/**
	 * 设置timeout
	 * @param timeout the timeout to set
	 */
	public void setTimeout(long timeout) {
		this.timeout = timeout;
	}

	/**
	 * 获取串行键
	 * @return the keyDeserializer
	 */
	public String getKeyDeserializer() {
		return keyDeserializer;
	}

	/**
	 * 设置串行键
	 * @param keyDeserializer the keyDeserializer to set
	 */
	public void setKeyDeserializer(String keyDeserializer) {
		this.keyDeserializer = keyDeserializer;
	}

	/**
	 * 获取串行值
	 * @return the valueDeserializer
	 */
	public String getValueDeserializer() {
		return valueDeserializer;
	}

	/**
	 * 设置串行值
	 * @param valueDeserializer the valueDeserializer to set
	 */
	public void setValueDeserializer(String valueDeserializer) {
		this.valueDeserializer = valueDeserializer;
	}

	/**
	 * 获取props
	 * @return the props
	 */
	public Properties getProps() {
		return props;
	}

	/**
	 * 设置props
	 * @param props the props to set
	 */
	public void setProps(Properties props) {
		this.props = props;
	}

	/**
	 * 获取主题
	 * @return the topics
	 */
	public List<String> getTopics() {
		return topics;
	}

	/**
	 * 设置主题
	 * @param topics the topics to set
	 */
	public void setTopics(List<String> topics) {
		this.topics = topics;
	}

	/**
	 * 获取消费者
	 * @return the consumer
	 */
	public KafkaConsumer<String, byte[]> getConsumer() {
		return consumer;
	}

	/**
	 * 设置消费者
	 * @param consumer the consumer to set
	 */
	public void setConsumer(KafkaConsumer<String, byte[]> consumer) {
		this.consumer = consumer;
	}

	/**
	 * 获取是否 正在运行
	 * @return the running
	 */
	public boolean isRunning() {
		return running;
	}

	/**
	 * 设置是否正在运行
	 * @param running the running to set
	 */
	public void setRunning(boolean running) {
		this.running = running;
	}

	/**
	 * 开始消费
	 */
	public synchronized void startConsumer() {
		initializeProps();

		this.consumer = new KafkaConsumer<>(this.props);
		running = true;

		logger.info("subscribe to topics: " + StringUtils.join(topics, ";"));

		this.consumer.subscribe(topics);

		List<ConsumerRecord<String, byte[]>> buffer = Lists.newArrayList();

		while (true) {
			if (!running) {
				break;
			}

			ConsumerRecords<String, byte[]> records = this.consumer.poll(this.timeout);

			for (ConsumerRecord<String, byte[]> record : records) {
				buffer.add(record);
			}

			if (buffer.size() >= this.batchSize) {
				//不可变的拷贝
				List<ConsumerRecord<String, byte[]>> bufferCopy=ImmutableList.copyOf(buffer);
				try {
					process(bufferCopy);
					if (!this.autoCommit) {
						consumer.commitSync();
					}
				} catch (Exception e) {
					logger.error("process error: " + e);
					processError(bufferCopy, e);
				} finally {
					buffer.clear();
				}
			}
		}

		this.consumer.close();
	}

	/**
	 * 初始化属性
	 */
	public void initializeProps() {
		if (null == props) {
			this.props = new Properties();
			this.props.put("bootstrap.servers", this.address);
			this.props.put("group.id", this.groupId);
			this.props.put("enable.auto.commit", ((this.autoCommit) ? ("true") : ("false")));
			this.props.put("key.deserializer", this.keyDeserializer);
			this.props.put("value.deserializer", this.valueDeserializer);
		}
	}
	/**
	 * 执行startConsumer方法
	 */
	@Override
	public void run() {
		startConsumer();
	}

	/**
	 * 进程
	 * @param buffer
	 * @throws Exception
	 */
	abstract public void process(List<ConsumerRecord<String, byte[]>> buffer) throws Exception;
	/**
	 * 进程错误
	 * @param buffer
	 * @param e
	 */
	abstract public void processError(List<ConsumerRecord<String, byte[]>> buffer, Exception e);
}
